package com.rickie.pvuv.domain;

import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

public class TupleSerializationSchemaV3 implements KafkaSerializationSchema<Tuple4<Long, Long, Long, Integer>> {
    private String topic;
    private ObjectMapper mapper;

    public TupleSerializationSchemaV3(String topic) {
        super();
        this.topic = topic;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(Tuple4<Long, Long, Long, Integer> longLongLongIntegerTuple4, @Nullable Long aLong) {
        byte[] b = null;
        if(mapper == null){
            mapper = new ObjectMapper();
        }
        try {
            b = mapper.writeValueAsBytes(longLongLongIntegerTuple4);
        } catch (JsonProcessingException e) {
            // 注意，在生产环境这是个非常危险的操作，
            // 过多的错误打印会严重影响系统性能，请根据生产环境情况做调整
            e.printStackTrace();
        }
        return new ProducerRecord<>(topic, b);
    }
}
